package sortFile

import (
	"awesomeProject/sortBigDataFile/convUtil"
	"bufio"
	"fmt"
	"io"
	"log"
	"os"
	"sort"
	"strconv"
	"sync"
	"time"
)

var (
	compareChan []chan int //每个tmp文件对应一个chan用于传输数据出来做对比
	waitGroup   sync.WaitGroup
	sortedBuf   chan string
)

func Run(resultPath, dataPath ,tmpDir string,dataCount int) {
	err := os.RemoveAll(tmpDir)
	if err != nil {
		fmt.Errorf("remove temp faile %v ",err)
	}
	start := time.Now()
	//go waiting()
	open, err := os.Open(dataPath)
	if err != nil {
		log.Fatalf("open file err%v", err)
	}
	fMap, err := splitAndSortFile(open,tmpDir, dataCount)
	fmt.Println("split file done!")
	if err != nil {
		fmt.Printf("split file err%v", err)
	}
	refile, _ := initResultFile(resultPath)
	writer := bufio.NewWriter(refile)
	compareChan = make([]chan int, len(fMap))
	for idx, f := range fMap {
		file, err := os.Open(f)
		if err != nil {
			log.Fatal("open file error ", err)
			break
		}
		reader := bufio.NewReader(file)
		compareChan[idx] = make(chan int, 50000)
		//读取子文件元素
		waitGroup.Add(1)
		go readTempFile(compareChan[idx], reader, idx)
	}
	waitGroup.Add(1)
	go writeSortedDataToResFile(writer, sortedBuf)
	waitGroup.Add(1)
	go writeIntoRetChan(compareChan)
	waitGroup.Wait()
	fmt.Println("sort done!")
	fmt.Println("sort time : ", time.Since(start))
}

type chanStatus struct {
	//是否当前chan获取数据
	isCurrentNext bool
	invalid       bool
}

type oneInts []*oneInt

func (o oneInts) Len() int {
	return len(o)
}

func (o oneInts) Less(i, j int) bool {
	return o[i].data <= o[j].data
}

func (o oneInts) Swap(i, j int) {
	o[i], o[j] = o[j], o[i]
}

type oneInt struct {
	data    int
	chanIdx int
}

func NewOneInt(data int, idx int) *oneInt {
	return &oneInt{
		data:    data,
		chanIdx: idx,
	}
}

func initResultFile(path string) (*os.File, error) {
	sortedBuf = make(chan string, 50000)
	err2 := os.Remove(path)
	if err2 != nil {
		if os.IsNotExist(err2) {
		} else if err2 != nil {
			err2 := os.Remove(path)
			fmt.Printf("remove retFile %v", err2)
		}
	}
	open, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0666)
	if err != nil {
		log.Fatal("create result file error", err)
	}
	return open, nil
}

func writeIntoRetChan(chs []chan int) {
	defer waitGroup.Done()
	defer close(sortedBuf)
	//每当一个chan关闭时就需要缩容
	comCap := 8
	compareS := make([]*oneInt, 0)
	//下一个去取数据的chan index 对应值为true
	putChansStatus := make([]*chanStatus, len(chs))
	for i := 0; i < len(putChansStatus); i++ {
		putChansStatus[i] = new(chanStatus)
	}
	//第一轮全部拿出一个
	for i := 0; i < len(chs); i++ {
		compareS = append(compareS, NewOneInt(<-chs[i], i))
	}
	//后面缺少哪个取哪个
	for {
		//判断len>cpus的情况
		if len(compareS) >= cpus {
			sortAndWrite(oneInts(compareS), putChansStatus)
			compareS = compareS[1:]
		}
		for idx, chStaus := range putChansStatus {
			if chStaus.isCurrentNext && !chStaus.invalid {
				v, ok := <-chs[idx]
				if ok {
					compareS = append(compareS, NewOneInt(v, idx))
				} else {
					//如果chan关闭了则将改chan置为invalid
					chStaus.invalid = true
					comCap--
				}
				putChansStatus[idx].isCurrentNext = false
				break
			}

		}
		//判断len<cpus的情况 ：有chan关闭了 此时缩容
		if len(compareS) < cpus {
			if len(compareS) > 1 {
				sortAndWrite(oneInts(compareS), putChansStatus)
				compareS = compareS[1:]
			} else {
				if len(compareS) == 1 {
					sortAndWrite(oneInts(compareS), putChansStatus)
					compareS = compareS[1:]
					continue
				}
				if isAllChanClose(putChansStatus){
					break
				}
				//for _, v := range compareS {
				//	s := strconv.Itoa(v.data) + "\n"
				//	sortedBuf <- s
				//}
				fmt.Println("break")
				break
			}
		}
	}
}

func isAllChanClose(chansStaus []*chanStatus) (ok bool) {
	ok = true
	for _, statu := range chansStaus {
		if !statu.invalid {
			ok = false
			return
		}
	}
	return
}

func readTempFile(ch chan<- int, reader *bufio.Reader, idx int) {
	for {
		n, err := reader.ReadBytes('\n')
		data := string(n)
		if err == io.EOF {
			break
		}
		if err != nil {
			break
		}
		toInt, err := convUtil.StrToInt(data)
		if err != nil {
			fmt.Println(idx, "-one number convert failed")
		}
		ch <- toInt
	}
	close(ch)
	waitGroup.Done()
	//fmt.Printf("--------done%v-------\n", idx)
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("r", r)
		}
	}()
}

func waiting() {
	fmt.Println("排序中")
	for {
		for _, r := range `—— \ | / ` {
			fmt.Printf("\r  %c    ", r)
			time.Sleep(10000)
		}
	}
}

func sortAndWrite(compareS oneInts, putChans []*chanStatus) {
	sort.Sort(compareS)
	s := strconv.Itoa(compareS[0].data) + "\n"
	//file.WriteString(s)
	sortedBuf <- s
	putChans[compareS[0].chanIdx].isCurrentNext = true
}

func writeSortedDataToResFile(refile *bufio.Writer, ch <-chan string) {
	for {
		v, ok := <-ch
		if ok {
			refile.WriteString(v)
		} else {
			fmt.Println("write to resChan ok")
			break
		}
	}
	refile.Flush()
	waitGroup.Done()
}
